package com.tdy.cdc.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.tdy.cdc.app.BaseApp;
import com.tdy.cdc.bean.ConfigBean;
import com.tdy.cdc.util.DateFormatUtil;
import com.tdy.cdc.util.FlinkSinkUtil;
import com.tdy.cdc.util.JDBCUtil;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.*;
import java.util.function.Function;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * @author NanHuang
 * @Date 2023/3/8
 */
public class DwdAppCommon extends BaseApp{
    public static void main(String[] args) {
        new DwdAppCommon().init(
                4001,
                "DwdAppCommon",
                2
        );
    }

    @Override
    public void invoke(StreamExecutionEnvironment env) {
        // 1 查询配置信息
        HashMap<String, ConfigBean> configInfos = readConfigInfo();
        // 2 获取所有要读取的topic
        ArrayList<String> topics = getAllTopics(configInfos);
        // 3 消费所有topic
        DataStreamSource<String> ds = super.readKafka(env, topics);
        // 4 处理字段映射
        SingleOutputStreamOperator<JSONObject> resultDs = processFields(ds, configInfos);
        // 5 写入Kafka
        resultDs.sinkTo(FlinkSinkUtil.getKafkaSink(false,"dwd_common_apps"));
    }


    private static void processThenStr(JSONObject obj, JSONObject dwdResult, String targetColumn, String thenStr) {
        if (thenStr.contains("=")) {
            // 表达式
            if (thenStr.contains("!")) {
                String[] thenWords = thenStr.split("!=");
                String flag = thenWords[1].replaceAll("\'", "").equals("false") ? "0" : "1";
                if (!flag.equals(obj.getString(thenWords[0]))) {
                    dwdResult.put(targetColumn,"1");
                }else {
                    dwdResult.put(targetColumn,"0");
                }
            } else {
                String[] thenWords = thenStr.split("=");
                String flag = thenWords[1].replaceAll("\'", "").equals("false") ? "0" : "1";
                if (flag.equals(obj.getString(thenWords[0]))) {
                    dwdResult.put(targetColumn,"1");
                }else {
                    dwdResult.put(targetColumn,"0");
                }
            }
        } else {
            // 字段 or 具体值
            if (thenStr.contains("\'")){
                dwdResult.put(targetColumn,thenStr.replaceAll("\'",""));
            } else {
                dwdResult.put(targetColumn,obj.get(thenStr));
            }
        }
    }

    private SingleOutputStreamOperator<JSONObject> processFields(DataStreamSource<String> ds, HashMap<String, ConfigBean> configInfos) {
        return ds.map(JSON::parseObject)
                .process(new ProcessFunction<JSONObject, JSONObject>() {
                    @Override
                    public void processElement(JSONObject obj, ProcessFunction<JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        String sourceTable = obj.getString("table_name");
                        if (sourceTable != null) {
                            ConfigBean configInfo = configInfos.get(sourceTable);
                            if (configInfo == null) {
                                // 如果不能直接获取通过"表名"获取配置信息，就通过"表名:xds_type"获取配置信息
                                configInfo = configInfos.get(sourceTable + ":" + obj.getString("xds_type"));
                            }
                            if (configInfo != null) {
                                // 配置信息不为null时，处理字段映射
                                String[] targetColumns = configInfo.getTargetColumns().split(",");
                                String[] sourceColumns = configInfo.getSourceColumns().split(",");
                                JSONObject dwdResult = new JSONObject();
                                dwdResult.put("table_name",configInfo.getTargetTable());
                                dwdResult.put("op",obj.getString("op"));
                                dwdResult.put("DW_DATE", DateFormatUtil.toYmdHms(System.currentTimeMillis()));
                                for (int i = 0; i < targetColumns.length; i++) {
                                    String targetColumn = targetColumns[i];
                                    String sourceColumn = sourceColumns[i];
                                    if (sourceColumn.contains("case") && sourceColumn.contains("when") && sourceColumn.contains("then") && sourceColumn.contains("end")){
                                        // when-then 字段
                                        String regex = "when\\s+(.*?)\\s+then\\s+(.*?)\\s+(?=when|else|$)";
                                        Pattern pattern = Pattern.compile(regex, Pattern.MULTILINE);
                                        Matcher matcher = pattern.matcher(sourceColumn);
                                        /*
                                            本项目中case-when的格式
                                                case
                                                when 表达式1 then 表达式1/字段1/具体值1
                                                when 表达式2 then 表达式2/字段2/具体值2
                                                else 表达式/字段/具体值
                                                end

                                            注意
                                                1.表达式中出现的判断字符目前只有 "=" 或 "!="
                                                2.then语句后的表达式，一般是比较字符串值，满足则标记为1，不满足标记为0
                                        */
                                        while (matcher.find()) {
                                            String whenStr = matcher.group(1).replaceAll(" ", "");
                                            String thenStr = matcher.group(2).replaceAll(" ", "");
                                            if (whenStr.contains("!")) {
                                                // 处理 "!=" 判断语句
                                                String[] words = whenStr.split("!=");
                                                if (!words[1].replaceAll("\'","").equals(obj.getString(words[0]))) {
                                                    // 处理then后的值
                                                    processThenStr(obj, dwdResult, targetColumn, thenStr);
                                                }
                                            } else {
                                                // 处理 "=" 判断语句
                                                String[] words = whenStr.split("=");
                                                if (words[1].replaceAll("\'","").equals(obj.getString(words[0]))) {
                                                    // 处理then后的值
                                                    processThenStr(obj, dwdResult, targetColumn, thenStr);
                                                }
                                            }
                                        }
                                        if (dwdResult.get(targetColumn) != null) {
                                            continue;
                                        }
                                        // else 字段
                                        String elseRegex = "else\\s+(.*)end";
                                        Pattern elsePattern = Pattern.compile(elseRegex, Pattern.MULTILINE);
                                        Matcher elseMatcher = elsePattern.matcher(sourceColumn);
                                        if (elseMatcher.find()) {
                                            String elseStr = elseMatcher.group(1);
                                            // 处理then后的值
                                            processThenStr(obj, dwdResult, targetColumn, elseStr);
                                        }

                                    } else {
                                        // 简单映射处理
                                        dwdResult.put(targetColumn,obj.get(sourceColumn));
                                    }
                                }
                                out.collect(dwdResult);
                            }
                        }
                    }
                });
    }

    private ArrayList<String> getAllTopics(HashMap<String, ConfigBean> configInfos) {
        List<String> topics = new ArrayList<>(configInfos.keySet());
        // 1 去掉不存在的Topic
        Iterator<String> iterator = topics.iterator();
        while (iterator.hasNext()){
            String t = iterator.next();
            if ("hdsd00_18_02".equals(t) || "hdsd00_18_14".equals(t) || "hdsd00_18_15".equals(t) || "hdsd00_06_02".equals(t) ) {
                iterator.remove();
            }
        }
        // 2 处理"表名:xds_type"
        return topics.stream().map(new Function<String, String>() {
            @Override
            public String apply(String s) {
                String[] words = s.split(":");
                return words[0];
            }
        }).distinct().collect(Collectors.toCollection(ArrayList::new));
    }

    private HashMap<String, ConfigBean> readConfigInfo() {
        try {
            Connection conn = JDBCUtil.getConfigConnection();
            String sql = "select * from dwd_config_copy1";
            PreparedStatement ps = conn.prepareStatement(sql);
            ResultSet resultSet = ps.executeQuery();
            HashMap<String, ConfigBean> result = new HashMap<>();
            while (resultSet.next()){
                ConfigBean config = new ConfigBean(
                        resultSet.getLong(1),
                        resultSet.getString(2),
                        resultSet.getString(3),
                        resultSet.getString(4),
                        resultSet.getString(5),
                        resultSet.getString(6)
                );
                String sourceTable = config.getSourceTable();
                String extend = config.getExtend();
                if (extend == null) {
                    result.put(sourceTable,config);
                }else {
                    // 有xds_type拓展信息的，将"表名:xds_type"作为key，配置信息对象作为value
                    JSONArray xdsType = JSONObject.parseObject(extend).getJSONArray("xds_type");
                    if (xdsType != null) {
                        for (Object o : xdsType) {
                            String xdsTypeValue = (String) o;
                            result.put(sourceTable + ":" + xdsTypeValue,config);
                        }
                    }
                }
            }
            ps.close();
            conn.close();
            return result;
        } catch (SQLException e) {
            throw new RuntimeException("连接MySQL中DWD配置表出错！");
        }
    }
}
